import ast
import json
import socket

# import pandas as pd
import pymysql
from pandas import to_numeric, NaT, to_datetime, NA,Timedelta,concat,notna
from datetime import datetime

def get_field_names(cursor, table_name,store_type):
    if  store_type == '京东' :
        columns=[]
        if table_name == 'ecapp_sjzlr':
            columns = ['date','category','page_views','visitors','paying_buyers','payment_conversion_rate','payment_suborders','payment_items','payment_amount','average_transaction_value','name']
        elif table_name == 'ecapp_lllydpr':
            columns = ['date', 'traffic_source', 'primary_source', 'secondary_source', 'tertiary_source', 'visitors',
                       'uv_value', 'average_transaction_value', 'add_to_cart_count', 'follow_shop_count','add_to_cart_rate',
                       'order_buyers_count','order_conversion_rate','order_amount','paying_buyers','payment_conversion_rate','payment_amount','name']
        elif table_name == 'ecapp_yxcj':
            columns = ['scene_id','scene_name','date', 'cost', 'impressions', 'clicks', 'avg_click_cost', 'cost_per_thousand_impressions',
                       'click_through_rate', 'total_sales_orders', 'total_sales_amount', 'total_cart_count','add_to_cart_rate','click_conversion_rate','roi','guided_potential_customer_rate','name']
        elif table_name == 'ecapp_hxgkr':
            columns = ['date','product_name','product_id', 'category', 'page_views', 'visitors', 'avg_page_views_per_visitor', 'avg_stay_duration','paying_buyers',
                       'payment_conversion_rate', 'order_buyers', 'order_items','payment_amount','average_transaction_value','cart_people','visit_to_cart_conversion_rate','cart_items','name']
        elif table_name == 'ecapp_lllyr':
            columns = ['date','product_id','traffic_source', 'primary_source', 'secondary_source', 'page_views', 'visitors',
                       'uv_value', 'average_transaction_value', 'cart_additions', 'order_buyers','paying_buyers','payment_conversion_rate','payment_amount','name']
        elif table_name == 'ecapp_ztr':
            columns = ['date','subject_name','product_id', 'subject_type', 'click_rate', 'spend', 'avg_click_spend',
                       'click_conversion_rate', 'roi', 'total_sales_amount', 'total_sales_count','clicks','category','exposure','name']
        return columns
    else:
        cursor.execute(f"SHOW COLUMNS FROM {table_name}")
        return [row[0] for row in cursor.fetchall()]

def get_matching_fields(cursor, table_name):
    model_name = table_name.split('_')[-1].upper()
    try:
        cursor.execute("SELECT matching_fields FROM ecapp_modelfieldmapping WHERE model_name = %s", (model_name,))
        result = cursor.fetchone()
        return json.loads(result[0]) if result else None  # 解析 JSON 字符串
    except BaseException as e:
        print(f"数据库查询错误: {e}")
        return None

def auto_data_to_mysql(file, table_name, storename,dburl,ztr_type=None,store_type=None,product_id=None,task_date=None,status_data=None):
    """
    自动获取的数据到 MySQL 表，如果有匹配条件，则更新记录，否则直接新增。
    :param file: 获取的数据 (DataFrame)
    :param table_name: 要导入数据的 MySQL 表名称
    :param storename: 要导入的店铺
    :param ztr_type: 主体报表的数据类型
    :param store_type: 店铺所属平台
    :param product_id: 产品ID或产品列表  京东天猫时为产品列表,其他是id
    :param task_date: 当前任务日期
    :param status_data: 待更新的任务状态
    """
    tableName = table_name.split('_')[1].upper()
    try:
        connection = pymysql.connect(
            host=dburl,
            user='root',  # 数据库用户名
            password='123',  # 数据库密码
            database='ecadb'  # 数据库名称
        )
        cursor = connection.cursor()
    except BaseException as e:
        cursor = None
        return
    # 数据清理：移除逗号并转换为正确的数据类型
    # 删除指定列
    df = file.drop(columns=['原二级场景ID', '原二级场景名字'], errors='ignore')

    # 获取表字段名
    field_names = get_field_names(cursor, table_name,store_type)
    if 'id' in field_names:
        field_names.remove('id')
    # 获取匹配字段
    matching_fields = get_matching_fields(cursor, table_name)

    # 按顺序替换 DataFrame 的列名
    try:
        df.columns = field_names[:len(df.columns)]
    except ValueError as e:
        print(f"Error: {e}")
        print(f"Error: {table_name}")
        print(f"Current DataFrame columns: {df.columns.tolist()}")
        print(f"Expected columns: {field_names}")
        raise  # 重新抛出异常，便于调试
    df = clean_data(df, table_name, storename, dburl,ztr_type)
    # ===================增加内容:数据总览和核心概况用当日的昨天数据覆盖昨天导入的当日数据=================开始
    if table_name == 'ecapp_sjzl':#核心概况不用,因为模型字段映射表没使用category
        copy_df = df.copy()
        # 获取昨天的日期
        yesterday = to_datetime('today') - Timedelta(days=1)

        # 提取 category 为 '昨天' 的数据
        yesterday_df = copy_df[copy_df['category'] == '昨天']

        # 修改昨天数据的 category 为 '今天'，date 修改为昨天的日期
        yesterday_df = yesterday_df.copy()
        yesterday_df['category'] = '今天'
        yesterday_df.loc[:, 'date'] = yesterday_df['date'].apply(lambda x: x.replace(year=yesterday.year, month=yesterday.month, day=yesterday.day))
        df = concat([copy_df, yesterday_df], ignore_index=True)
    # ===================增加内容:数据总览和核心概况用当日的昨天数据覆盖昨天导入的当日数据=================结束
    # 根据 field_names 过滤 DataFrame
    filtered_data = df[field_names].to_dict(orient='records')
    try:
        for item in filtered_data:
            obj_data = {key: value for key, value in item.items() if key in field_names}

            if matching_fields:
                # 有匹配条件，使用 update 或 insert
                filter_kwargs = {field: obj_data[field] for field in matching_fields if field in obj_data}
                filter_conditions = ' AND '.join([f"{key} = %s" for key in filter_kwargs])
                filter_values = list(filter_kwargs.values())

                # 查找现有记录
                cursor.execute(f"SELECT * FROM {table_name} WHERE {filter_conditions}", filter_values)
                existing_record = cursor.fetchone()

                if existing_record:
                    # 更新记录
                    update_columns = ', '.join([f"{key} = %s" for key in obj_data])
                    update_values = list(obj_data.values())
                    update_query = f"UPDATE {table_name} SET {update_columns} WHERE {filter_conditions}"
                    cursor.execute(update_query, update_values + filter_values)
                else:
                    # 插入新记录
                    columns = ', '.join(obj_data.keys())
                    placeholders = ', '.join(['%s'] * len(obj_data))
                    insert_query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
                    cursor.execute(insert_query, list(obj_data.values()))
            else:
                # 无匹配条件，直接插入新对象
                columns = ', '.join(obj_data.keys())
                placeholders = ', '.join(['%s'] * len(obj_data))
                insert_query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
                cursor.execute(insert_query, list(obj_data.values()))
        connection.commit()  # 提交事务
        if product_id is not None:
            if (store_type == '京东' or store_type == '天猫' ) and tableName == 'HXGKR':
                for proId in product_id:
                    status_data[storename][proId][tableName] = True
            else:
                status_data[storename][product_id][tableName]  = True
        else:
            if tableName =='关键词推广':
                tableName = 'ZTR_GJ'
            elif tableName =='人群推广':
                tableName = 'ZTR_RQ'
            elif tableName == '货品运营':
                tableName = 'ZTR_HP'
            elif tableName == '全站推广':
                tableName = 'ZTR_QZ'
            status_data[storename][tableName] = True
        up_task_status(cursor, task_date, status_data)
    except BaseException as e:
        print(f"数据写入数据库表:{table_name}失败",str(e))
        up_task_status(cursor, task_date, status_data, error_reason=str(e))
        connection.rollback()  # 回滚操作
    finally:
        cursor.close()
        connection.close()


def get_field_types(table_name,dburl):
    connection = pymysql.connect(
        host=dburl,
        user='root',
        password='123',
        database='ecadb'
    )
    cursor = connection.cursor()

    cursor.execute(f"""
        SELECT COLUMN_NAME, DATA_TYPE 
        FROM INFORMATION_SCHEMA.COLUMNS 
        WHERE TABLE_NAME = '{table_name}' 
          AND TABLE_SCHEMA = 'ecadb';  # 替换为你的数据库名
    """)

    fields = cursor.fetchall()
    decimal_fields = []
    float_fields = []
    integer_fields = []
    date_fields = []

    for column_name, data_type in fields:
        if data_type in ['decimal', 'numeric']:
            decimal_fields.append(column_name)
        elif data_type in ['float', 'double']:
            float_fields.append(column_name)
        elif data_type in ['int', 'integer']:
            integer_fields.append(column_name)
        elif data_type in ['date', 'datetime', 'timestamp']:
            date_fields.append(column_name)

    cursor.close()
    connection.close()
    return decimal_fields, float_fields, integer_fields, date_fields

def clean_data(df, table_name, storename,dburl,ztr_type=None):
    """
    清理 CSV 数据：移除数字中的逗号并尝试转换为正确的数据类型。
    对于数值字段，将 NaN 替换为 0.00，对于其他字段，NaN 替换为空字符串。
    :param df: 需要清理的 DataFrame
    :return: 清理后的 DataFrame
    """
    ev_df = df.copy()
    # 假设根据表结构定义数据类型
    decimal_fields, float_fields, integer_fields, date_fields = get_field_types(table_name,dburl)

    current_time = datetime.now()
    current_hour = current_time.hour
    if table_name.endswith('r'):
        midnight_time = current_time.replace(hour=0, minute=0, second=0, microsecond=0)
        # 日数据的date默认都是0点
        current_hour = midnight_time.hour
        if table_name == 'ecapp_sjzlr':
            # 只保留本店数据
            ev_df = ev_df[ev_df['category'] == '本店']

    for column in ev_df.columns:
        # 对所有数值类型的列和对象类型的列处理 NaN 值
        if column in decimal_fields or column in float_fields or column in integer_fields:
            ev_df[column] = ev_df[column].replace('-', NA)
            ev_df[column] = ev_df[column].astype(str).str.replace(',', '', regex=False)  # 移除逗号
            ev_df[column] = to_numeric(ev_df[column], errors='coerce')  # 转换为数值
            ev_df[column] = ev_df[column].fillna(0.00)  # 将 NaN 替换为 0.00
        elif column in date_fields:
            def process_date(x):
                try:
                    if isinstance(x, str) and len(x.strip()) == 10:
                        parsed_date = to_datetime(x, errors='coerce')
                        return parsed_date.replace(hour=current_hour, minute=0, second=0)
                    else:
                        return to_datetime(x, errors='coerce')
                except Exception:
                    return NaT  # 返回 NaT

            ev_df[column] = ev_df[column].apply(process_date)
        else:
            # if ev_df[column].dtype == 'object':
            ev_df[column] = to_numeric(ev_df[column], errors='ignore')  # 尝试转换为数值
                # df.loc[:, column] = pd.to_numeric(df.loc[:, column], errors='ignore')#上面的写法会报警告 2025-01-10
            ev_df[column] = ev_df[column].fillna('')  # 字符串类型字段的 NaN 替换为空字符串

    # 添加新列
    if table_name == 'ecapp_ztr' and ztr_type is not None:
        ev_df['category'] = ztr_type
    ev_df['name'] = storename
    return ev_df

def get_user_ip():
    """
    获取用户的 IP 地址，这是本地地址。
    """
    # 例如通过 socket 获取本机的 IP 地址
    return socket.gethostbyname(socket.gethostname())

def up_task_status(cursor, task_date, status_data, error_reason=None):
    """
    向 download_task 表插入或更新数据,记录下载数据任务状态
    :param cursor: 数据库游标对象
    :param task_date: str, 日期字符串，作为主键
    :param status_data: dict, 任务状态数据，包含任务名称及其状态（True/False）
    :param error_reason: str, 错误原因，默认 None
    """
    try:
        status_data_json = json.dumps(status_data, ensure_ascii=False)
        ip_address = get_user_ip()
        # 插入或更新任务
        upsert_sql = """
        INSERT INTO download_task (task_name, task_status, end_time, error_reason,ip_address)
        VALUES (%s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE 
        task_status = VALUES(task_status),
        end_time = VALUES(end_time),
        error_reason = VALUES(error_reason),
        ip_address = VALUES(ip_address)
        """
        cursor.execute(upsert_sql, (task_date, status_data_json, datetime.now(), error_reason, ip_address))
        cursor.connection.commit()
    except Exception as e:
        print(f"Error in upsert_download_task: {e}")
        raise

def only_up_task_status(dburl,task_date, status_data, error_reason=None):
    """
    单独更新 download_task下载数据任务状态
    :param task_date: str, 日期字符串，作为主键
    :param status_data: dict, 任务状态数据，包含任务名称及其状态（True/False）
    :param error_reason: str, 错误原因，默认 None
    """
    try:
        connection = pymysql.connect(
            host=dburl,
            user='root',  # 数据库用户名
            password='123',  # 数据库密码
            database='ecadb'  # 数据库名称
        )
        cursor = connection.cursor()
    except BaseException as e:
        cursor = None
        return
    try:
        status_data_json = json.dumps(status_data, ensure_ascii=False)
        ip_address = get_user_ip()
        # 插入或更新任务
        upsert_sql = """
        INSERT INTO download_task (task_name, task_status, end_time, error_reason,ip_address)
        VALUES (%s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE 
        task_status = VALUES(task_status),
        end_time = VALUES(end_time),
        error_reason = VALUES(error_reason),
        ip_address = VALUES(ip_address)
        """
        cursor.execute(upsert_sql, (task_date, status_data_json, datetime.now(), error_reason, ip_address))
        cursor.connection.commit()
    except Exception as e:
        print(f"Error in upsert_download_task: {e}")
        raise
    finally:
        cursor.close()
        connection.close()

def query_task_status(cursor, task_date):
    """
    查询指定日期的任务状态，并判断所有任务状态是否都为 True。
    :param cursor: 数据库游标对象
    :param task_date: str, 日期字符串（主键）
    :return: dict 或 None, 查询结果，包含任务状态和是否全部为 True 的标志
    """
    try:
        ip_address = get_user_ip()
        select_sql = "SELECT * FROM download_task WHERE task_name = %s and ip_address = %s"
        cursor.execute(select_sql, (task_date,ip_address))
        result = cursor.fetchone()
        if result:
            # 获取列名并转为字典
            columns = [desc[0] for desc in cursor.description]
            result_dict = dict(zip(columns, result))
            task_status = json.loads(result_dict['task_status'])
            return task_status
        return None
    except Exception as e:
        print(f"查询任务状态错误: {e}")
        raise

